package com.cirnoworks.uhs;

import java.io.Closeable;
import java.io.IOException;
import java.net.InetSocketAddress;
import java.net.SocketAddress;
import java.nio.ByteBuffer;
import java.nio.channels.ClosedChannelException;
import java.nio.channels.DatagramChannel;
import java.nio.channels.SelectionKey;
import java.nio.channels.Selector;
import java.util.ArrayDeque;
import java.util.Deque;
import java.util.Random;

import org.slf4j.Logger;
import org.slf4j.LoggerFactory;

public abstract class Link implements Closeable {
	private final static Logger LOG = LoggerFactory.getLogger(Link.class);
	private Deque<Packet> writeBuf = new ArrayDeque<Packet>();
	protected final DatagramChannel channel;
	private ByteBuffer buf = ByteBuffer.allocateDirect(65536);
	private final LinkManager manager;
	protected final ThreadLocal<Random> rand = new ThreadLocal<Random>() {

		@Override
		protected Random initialValue() {
			return new Random();
		}

	};

	public Link(Selector selector, LinkManager manager, int port)
			throws IOException {
		this.manager = manager;
		channel = DatagramChannel.open();
		if (port > 0) {
			channel.socket().bind(new InetSocketAddress(port));
		}
		channel.configureBlocking(false);
		channel.register(selector, SelectionKey.OP_READ, this);
		manager.registerLink(this);
	}

	public Link(Selector selector, LinkManager manager) throws IOException {
		this(selector, manager, -1);
	}

	public void sendData(Selector selector, SocketAddress address,
			ByteBuffer data) throws ClosedChannelException {
		if (address == null || data == null) {
			throw new NullPointerException();
		}
		channel.register(selector,
				SelectionKey.OP_READ | SelectionKey.OP_WRITE, this);
		writeBuf.add(Packet.create(address, data));
	}

	public void onDataReceived(Selector selector) throws IOException {
		// LOG.info("Data got by " + this);
		buf.clear();
		beforeReceiveData(buf);
		SocketAddress address = channel.receive(buf);
		buf.flip();
		// LOG.info("Data got by " + this + " from " + address);
		receiveData(selector, address, buf);
	}

	public void beforeReceiveData(ByteBuffer buf) {

	}

	public void onWritable(Selector selector) throws IOException {
		if (writeBuf.isEmpty()) {
			channel.register(selector, SelectionKey.OP_READ, this);
		} else {
			Packet packet = writeBuf.poll();
			if (packet != null) {
				packet.sendTo(channel);
			}
		}
	}

	protected abstract void receiveData(Selector selector, SocketAddress from,
			ByteBuffer buf) throws IOException;

	public final void close() throws IOException {
		try {
			onClose();
		} catch (Exception e) {
			LOG.error("Error closing link " + this, e);
		} finally {
			manager.onLinkClosed(this);
			channel.close();
		}
	}

	public abstract void onClose();
}
